
#include "processhub.h"
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>
#include <signal.h>
#include <stdio.h>
#include <sys/sysinfo.h>
#include <algorithm>

#include <QApplication>
#include <QString>

#include "../core/browser.h"
#include "../activite/webactivite.h"
#include "../activite/webcollector.h"
#include "../custom/webspidercustom.h"
#include "../../third/log4qt/src/log4qt/logmanager.h"
#include "../../third/log4qt/src/log4qt/logger.h"
#include "../../third/log4qt/src/log4qt/basicconfigurator.h"
#include "../../third/log4qt/src/log4qt/propertyconfigurator.h"
#include "../../third/log4qt/src/log4qt/helpers/properties.h"
#include <google/coredumper.h>

namespace ProcessRouter {

ProcessHub* ProcessHub::m_instance = NULL;
ManagerInfo* ProcessHub::m_managerInfo = NULL;

// signal quit
void signal_quit(int no)
{
    ProcessHub::instance()->close();

    if (no == SIGSEGV || no == SIGABRT) {
        if( WriteCoreDump("/tmp/spiderkit.coredump") ==0 )
            printf("success: WriteCoreDump to coredump-pid file.\n");
        else
            printf("failure: WriteCoreDump to coredump-pid file.\n");
    }

    for(int i = 0; i < 257; i ++) {
        printf("end\n");
    }

    exit(0);
}

ProcessHub::ProcessHub()
    : m_task(this, &ProcessHub::taskLoop)
    , m_bindEnd(false)
    , m_channel(NULL)
    , m_zmqchannel(NULL)
    , m_iothread(NULL)
    , m_zkthread(NULL)
{
    signal(SIGQUIT, signal_quit);
    signal(SIGINT, signal_quit);
    signal(SIGSEGV, signal_quit);
    signal(SIGABRT, signal_quit);

    m_task.reg();

    rooftop_add_event(CHILDPROCESSERROREVENT); 
    Rooftop::EventManager::getInstance()->addEventCallback(CHILDPROCESSERROREVENT,
            boost::bind(&ProcessHub::clearChild, this, _1));

    std::string type_name = ProcessRouter::IPCCommandProxy::descriptor()->full_name();
    rooftop_add_event(type_name.c_str()); 
    Rooftop::EventManager::getInstance()->addEventCallback(type_name.c_str(),
            boost::bind(&ProcessHub::childUpdateProxy, this, _1));

}

ProcessHub::~ProcessHub()
{
    rooftop_remove_event(MESSAGEEVENT); 
}

ProcessHub* ProcessHub::instance(std::string path)
{
    if (!m_instance) {
        rooftop_add_event(MESSAGEEVENT); 
        m_managerInfo = new ManagerInfo();

        if (m_managerInfo->initFromLocalFile(path) == -1) {
            fprintf(stderr, "Init configuer error !\n");
            exit(-1);
        }

        m_instance = new ProcessHub();
    }

    return m_instance;
}

ManagerInfo* ProcessHub::info()
{
    return m_managerInfo;
}

std::string ProcessHub::serializePageInfo()
{
    return m_managerInfo->serializePageInfo();
}

void ProcessHub::bind()
{
    if (m_bindEnd)
        return;

    if (!m_iothread) {
        m_iothread = new IOThread();
        Rooftop::EventManager::getInstance()->addEventCallback(MESSAGEEVENT,
                boost::bind(&ProcessHub::onRecvPackage, m_instance, _1));
    }

    m_zmqchannel = new ZmqChannel(m_iothread);
    m_zmqchannel->setProtocol(m_managerInfo->protocol());
    ZmqMessageHandle *zmqhandle = new ZmqMessageHandle();
    m_zmqchannel->open();
    m_zmqchannel->setClient(zmqhandle);

    m_zkthread = new ZKThread();
    m_zkthread->start();

    m_bindEnd = true;
}

void ProcessHub::close()
{
    m_bindEnd = false;
    m_task.unreg();
    m_zkthread->stop();

    usleep(500000);

    if (m_zkthread)
        delete m_zkthread;

    ZmqMessageHandle *zmqhandle = (ZmqMessageHandle*)m_zmqchannel->client();
    if (zmqhandle)
        zmqhandle->doExit();

    m_zmqchannel->destroy();
 
    std::map<int, FifoChannel*>::iterator it = m_fifos.begin();
    for ( ; it != m_fifos.end(); it ++)
        sendKill(it->second);

    usleep(500000);
}

void ProcessHub::sendKill(FifoChannel* fc)
{
    if (!fc)
        return;


    ProcessRouter::IPCCommandKill *command = new ProcessRouter::IPCCommandKill();
    command->set_signo(SIGKILL);

    ProcessRouter::MessagePackage package;
    std::string type_name = ProcessRouter::IPCCommandKill::descriptor()->full_name();
    package.set_name(type_name);
    package.set_allocated_commandkill(command);

    std::string kmess;
    package.SerializeToString(&kmess);
    CMessage message(kmess);
    fc->send(message);
}

int ProcessHub::createProcess(bool isStatic)
{
    if (m_managerInfo->mempercent() <= SYSTEM_MEMORY_LIMIT_PERCENT)
        return -1;

    FifoChannel *channel = new FifoChannel(m_iothread);
    channel->open();
    
    if (m_managerInfo->isEnableProxyPool())
        m_proxy = m_managerInfo->getProxyFromPool();

    int pid = channel->cfork();
    if (pid == 0) {
        m_task.unreg();

        m_channel = channel;
        m_isStatic = isStatic;

        subProcessMain();
    } else if (pid > 0) {

        ManagerMessageHandle *messageHandle = new ManagerMessageHandle();
        channel->forkInit(messageHandle);

        m_fifos.insert(std::pair<int, FifoChannel*>(pid, channel));

        if (isStatic)
            m_staticProcess.push_back(pid);
        else
            m_dynamicProcess.push_back(pid);

        return 0;
    } else {
        perror("Sub process field");
        return -1;
    }
}

bool ProcessHub::clearChild(Rooftop::EventBase &e)
{
    ChildProceeErrorEvent *event = static_cast<ChildProceeErrorEvent*>(&e);
    int cpid = event->cpid();

    std::cout << "clear child : " << cpid << std::endl;

    clearFifoChannel(cpid);

    m_staticProcess.remove(cpid);
    m_dynamicProcess.remove(cpid);

    return true;
}

bool ProcessHub::childUpdateProxy(Rooftop::EventBase &e)
{
    PackageEvent *event = static_cast<PackageEvent*>(&e);
    if (!event->package().has_commandproxy())
        return false;

    IPCCommandProxy *command = event->package().mutable_commandproxy();

    std::string proxy = m_managerInfo->getProxyFromPool();

    int pos = proxy.find(":");
    if(pos == -1)
        return false;

    std::string host = proxy.substr(0, pos);
    std::string portStr = proxy.substr(pos+1, proxy.size());
    int port = atoi(portStr.c_str());
    command->set_host(host);
    command->set_port(port);

    FifoChannel *channel = getChannel(command->pid());
    if (!channel)
        return false;

    std::string newcontent;
    event->package().SerializeToString(&newcontent);

    ProcessRouter::CMessage message(newcontent);
    channel->send(message);

    return true;
}

FifoChannel* ProcessHub::getChannel(int aimpid)
{
    if (m_managerInfo->inNeedDestroyProcess(aimpid))
        return NULL;

    std::map<int, FifoChannel*>::iterator it = m_fifos.find(aimpid);
    if (it == m_fifos.end())
        return NULL;

    return it->second;
}

void ProcessHub::subProcessMain()
{
    /*
    // clean main process res
    for (std::map<int, FifoChannel*>::iterator it=m_fifos.begin(); it!=m_fifos.end(); it++) {
        if (it->second)
            delete it->second;
    }
    m_fifos.clear();
    */

    // create webkit process
    WebkitMessageHandle *messageHandle = new WebkitMessageHandle();
    m_channel->forkInit(messageHandle);
    m_iothread->start();

    char* argv = NULL;
    int argc = 0;

    QApplication app(argc, &argv);

    app.setApplicationName(PRODUCTNAME);

    Browser::instance();

    WebCollector *collector = WebCollector::instance();

    if (m_managerInfo->configuer()->isdebug())
        collector->setDebug();

    if (m_isStatic)
        collector->setStatic();

    if (m_managerInfo->configuer()->exitworkcount() > 0)
        collector->setExitWorkCount(m_managerInfo->configuer()->exitworkcount());

    //Log4Qt::BasicConfigurator::configure();

    Log4Qt::Properties logConfig;

    if (m_managerInfo->configuer()->isdebug())
        logConfig.setProperty("log4j.rootCategory", "debug,R");
    else
        logConfig.setProperty("log4j.rootCategory", "info,R");

    logConfig.setProperty("log4j.appender.stdout", "org.apache.log4j.ConsoleAppender");
    logConfig.setProperty("log4j.appender.stdout.layout", "org.apache.log4j.PatternLayout");
    logConfig.setProperty("log4j.appender.stdout.layout.ConversionPattern", "\%d{yyyy-MM-dd HH:mm:ss} [\%c] [\%p] - \%m\%n");
    logConfig.setProperty("log4j.appender.R", "org.apache.log4j.DailyRollingFileAppender");
    logConfig.setProperty("log4j.appender.R.layout", "org.apache.log4j.PatternLayout");
    logConfig.setProperty("log4j.appender.R.layout.ConversionPattern", "\%d{yyyy-MM-dd HH:mm:ss} [\%c] [\%p] - \%m\%n");

    QString filePath = "logs/spiderkit." + QString::number(getpid()) + ".log";
    //QString filePath = "/dev/null";
    logConfig.setProperty("log4j.appender.R.File", filePath);

    Log4Qt::LogManager::setHandleQtMessages(true);
    Log4Qt::PropertyConfigurator::configure(logConfig);

    WebSpider *spider = new WebSpider((Channel*)m_channel);
    spider->setName(SPIDER_NAME);

    if (m_managerInfo->isEnableProxyPool()) {

        int pos = m_proxy.find(":");
        if(pos != -1) {
            std::string host = m_proxy.substr(0, pos);
            std::string portStr = m_proxy.substr(pos+1, m_proxy.size());
            int port = atoi(portStr.c_str());
            collector->setProxy("http", "", "", host.c_str(), port);
        }
    } else if (m_managerInfo->isEnableProxy()) {
        collector->setProxy(
                m_managerInfo->proxyType().c_str(),
                m_managerInfo->proxyAuthUser().c_str(),
                m_managerInfo->proxyAuthPass().c_str(),
                m_managerInfo->proxyHost().c_str(),
                m_managerInfo->proxyPort());
    }

    collector->enableProxy();

    collector->insert(spider);

    app.exec();
}

void ProcessHub::clearFifoChannel(int cpid)
{
    std::map<int, FifoChannel*>::iterator it = m_fifos.find(cpid);
    if (it == m_fifos.end())
        return;

    if (it->second) {
        it->second->destroy();
        delete it->second;
        m_fifos.erase(it);
    }
}

bool ProcessHub::onRecvPackage(Rooftop::EventBase &e)
{
    CMessageEvent *event = static_cast<CMessageEvent*>(&e);
    if (event->type() == CMessageEvent::FIFOTYPE) {
        m_filter.dispatch(event->message(), false);
        if (m_managerInfo->inNeedDestroyProcess(event->fromPid())) {

            bool needKill = false;
            ZmqMessageHandle *zmqhandle = (ZmqMessageHandle*)m_zmqchannel->client();
            if (zmqhandle)
                if (!zmqhandle->hasCpidRequest(event->fromPid()))
                    needKill = true;
            else
                needKill = true;

            m_managerInfo->removeFromNeedDestroyProcessList(event->fromPid());
            if (needKill) {
                std::map<int, FifoChannel*>::iterator it = m_fifos.find(event->fromPid());
                if (it != m_fifos.end()) {
                    sendKill(it->second);
                    std::cout << "kill : " << event->fromPid() << std::endl;
                }
            }
        }
    } else if (event->type() == CMessageEvent::ZMQTYPE) {
        m_filter.dispatch(event->message(), true);
        ZmqMessageHandle *zmqhandle = (ZmqMessageHandle*)m_zmqchannel->client();
        if (zmqhandle)
            info()->setConnectSize(zmqhandle->connectSize());
    }
}

void ProcessHub::taskLoop(Rooftop::Task<ProcessHub> *)
{
    if (m_managerInfo && m_bindEnd) {

        m_managerInfo->setCoreSize(m_staticProcess.size() + m_dynamicProcess.size());

        if (m_managerInfo->mempercent() > SYSTEM_MEMORY_LIMIT_PERCENT) {
            int staticWkCount = m_managerInfo->configuer()->staticcorecount();
            if (m_staticProcess.size() < staticWkCount) {
                int needCreate = staticWkCount - m_staticProcess.size();
                for (int i = 0; i < needCreate; i ++)
                    createProcess(true);
            }

            int dynamicWkCount = m_managerInfo->configuer()->dynamiccorecount();
            if (m_dynamicProcess.size() < dynamicWkCount) {
                int needCreate = dynamicWkCount - m_dynamicProcess.size();
                for (int i = 0; i < needCreate; i ++)
                    createProcess(false);
            }
        }
    }
}

}

